/*
 * Copyright 2023-2024 the original author or authors.
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package com.icodening.collapse.core;

import java.util.Collection;
import java.util.Iterator;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executor;
import java.util.stream.Collectors;

/**
 * @author icodening
 * @date 2023.05.14
 */
public abstract class AbstractCollapseExecutor<INPUT, OUTPUT, BATCH_OUTPUT> implements CollapseExecutor<INPUT, OUTPUT>, NamedCollapseExecutor {

    private ListeningCollector collector;

    private String name = this.getClass().getSimpleName();

    private InputGrouper<INPUT> inputGrouper = EqualsInputGrouper.getInstance();

    public AbstractCollapseExecutor() {
        this(new SuspendableCollector());
    }

    public AbstractCollapseExecutor(ListeningCollector collector) {
        this.collector = Objects.requireNonNull(collector, "collector must be not null.");
        this.setCollector(collector);
    }

    /**
     * 1、移除原有收集器中 执行器对应的数据（移除原有收集器中，对应的折叠执行器 对应的数据（key,收集器监听器-函数式接口-实现为折叠执行逻辑））
     * 2、覆盖collector
     * 3、为收集器添加监听器，监听器逻辑为行为参数（为收集器添加新的折叠执行器 以及 对应的折叠执行逻辑）
     *
     * @param collector
     */
    public void setCollector(ListeningCollector collector) {
        ListeningCollector old = this.collector;
        if (old != null) {
            old.removeListener(this);
        }
        this.collector = collector;
        this.collector.addListener(this, this::collapseExecute);
    }

    public void setInputGrouper(InputGrouper<INPUT> inputGrouper) {
        this.inputGrouper = Objects.requireNonNull(inputGrouper, "inputGrouper must be not null.");
    }

    public InputGrouper<INPUT> getInputGrouper() {
        return inputGrouper;
    }

    protected ListeningCollector getCollector() {
        return collector;
    }

    public void setName(String name) {
        this.name = name;
    }

    @Override
    public String getName() {
        return name;
    }

    /**
     * 合并执行器执行逻辑
     * 收集器、执行器的所有入口
     *
     * @param input
     * @return
     * @throws Throwable
     */
    @Override
    @SuppressWarnings("unchecked")
    public OUTPUT execute(INPUT input) throws Throwable {
        //获取回调线程池
        Executor callbackExecutor = getCallbackExecutor();
        //创建绑定
        Bundle<INPUT, OUTPUT> bundle = createBundle(input, callbackExecutor, new CompletableFuture<>());
        //获取可监听的收集器
        ListeningCollector collector = Objects.requireNonNull(this.collector, "collector must be not null.");
        //将绑定 放入收集器 队列
        collector.enqueue((Bundle<Object, Object>) bundle);
        //将bundle转换为OUYPUT结果返回
        return returning(bundle);
    }

    /**
     * 将结果返回
     *
     * @param bundle
     * @return
     * @throws Throwable
     */
    protected abstract OUTPUT returning(Bundle<INPUT, OUTPUT> bundle) throws Throwable;

    /**
     * 回去回调执行器线程池
     *
     * @return
     */
    protected abstract Executor getCallbackExecutor();

    /**
     * 将结合输入包装对象 执行 批量接口 返回批量输出
     * <p>
     * 用于收集完之后，收到收集完成事件后，执行的逻辑
     *
     * @param inputs
     * @return
     * @throws Throwable
     */
    protected abstract BATCH_OUTPUT doExecute(Collection<Input<INPUT>> inputs) throws Throwable;

    /**
     * 将批量结果对象 转换为 集合的Bunlde
     *
     * @param batchOutput
     * @param bundles
     */
    protected abstract void bindingOutput(BATCH_OUTPUT batchOutput, List<Bundle<INPUT, OUTPUT>> bundles);

    /**
     * 绑定关系
     *
     * @param input             输入
     * @param executor          out - 回调执行器
     * @param completableFuture cf
     * @return bundle
     */
    private Bundle<INPUT, OUTPUT> createBundle(INPUT input, Executor executor, CompletableFuture<OUTPUT> completableFuture) {
        return new Bundle<>(this, input, executor, completableFuture);
    }

    /**
     * 合并执行逻辑
     * 单个收集器，收集完成后，执行的逻辑
     *
     * @param bundles
     */
    @SuppressWarnings("unchecked")
    private void collapseExecute(Collection<Bundle<Object, Object>> bundles) {
        //将bundle 转化为 Input对象
        List<Input<INPUT>> inputsSnapshot = bundles.stream().map(bundle -> (Input<INPUT>) new Input<>(bundle.getInput(), bundle)).collect(Collectors.toList());
        //对收集的Input集合按照分组逻辑进行分组
        Collection<Collection<Input<INPUT>>> groups = inputGrouper.grouping(inputsSnapshot);
        //对分组进行处理
        for (Collection<Input<INPUT>> group : groups) {
            if (group == null) {
                continue;
            }
            Iterator<Input<INPUT>> iterator = group.iterator();
            //分组后，没有元素，继续
            if (!iterator.hasNext()) {
                continue;
            }
            //分组后，有元素，执行下面逻辑
            //获取输入对象的bundle
            List<Bundle<INPUT, OUTPUT>> bundleGroup = group.stream().map(input -> (Bundle<INPUT, OUTPUT>) input.getBundle()).collect(Collectors.toList());
            //获取输入包装对象
            Input<INPUT> input = iterator.next();
            //使用bundle的回调线程池，执行相关折叠执行器的执行逻辑
            input.getBundle().getCallbackExecutor().execute(() -> {
                try {
                    //将收集的Input组作为入参，执行器执行相关逻辑
                    BATCH_OUTPUT batchOutput = this.doExecute(group);
                    //将批量结果，转换为对应的bundle中的output
                    this.bindingOutput(batchOutput, bundleGroup);
                } catch (Throwable e) {
                    for (Bundle<?, ?> bundle : bundleGroup) {
                        //将异常绑定到bundle中的异常上
                        bundle.bindOutput(null, e);
                    }
                }
            });
        }
    }

    @Override
    public final boolean equals(Object o) {
        return super.equals(o);
    }

    @Override
    public final int hashCode() {
        return super.hashCode();
    }
}
